package realtime.app.dim;


import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import realtime.bean.TableProcess;
import realtime.func.DimMyprocessFunction;
import realtime.func.DimPhenixSinkFunction;
import realtime.util.MyKafkaUtil;

/**
 * @author MengX
 * @create 2023/3/24 13:25:19
 */
public class BaseDimApp {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //TODO 状态后端设置
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/BaseDimApp/220926");
//
//        //设置HDFS用户信息
//        System.setProperty("HADOOP_USER_NAME", "root");
        //获取数据(主流）
        String topic = "topic_db";
        String groupId = "base_dim_app";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    try{
                        JSONObject jsonv = JSONObject.parseObject(value);
                        out.collect(jsonv);
                    }catch (JSONException e){
                        System.out.println("数据无法转化JSON，脏数据："+value);
                    }
                }
            }
        });
        //获取配置流
        MySqlSource<String> build = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("gmall_config")
                .tableList("gmall_config.table_process")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();
        DataStreamSource<String> cdcSource = env.fromSource(build, WatermarkStrategy.noWatermarks(), "cdcSource");
        cdcSource.print("cdcSource>>>>>>>>>>>>>>>>>>>>>>>");
        //转换广播流
        MapStateDescriptor<String, TableProcess> broadCastState = new MapStateDescriptor<>("tableProcess", String.class, TableProcess.class);
        BroadcastStream<String> broadcast = cdcSource.broadcast(broadCastState);
        //链接双流
        BroadcastConnectedStream<JSONObject, String> connect = jsonDS.connect(broadcast);
        SingleOutputStreamOperator<JSONObject> processDS = connect.process(new DimMyprocessFunction(broadCastState));
        processDS.print("processDS>>>>>>>>>>>>>>>>>>>>");
        processDS.addSink(new DimPhenixSinkFunction());

        env.execute("BaseDimApp");
    }
}
